package com.bigknow.appstore.server.rpc;

import com.bigknow.appstore.common.rpc.*;
import com.bigknow.appstore.common.rpc.client.Constants;
import com.bigknow.net.minidev.json.JSONValue;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.exception.ExceptionUtils;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

public class RPCSession {

    private RPCServer server;

    private ChannelHandlerContext ctx;

    private String sysId;

    private byte[] msgIdBuf;

    private byte[] clzNameBuf;

    private Map<String, RPCResponse> responseMap = new HashMap<String, RPCResponse>();

    private static AtomicLong idGen = new AtomicLong(System.currentTimeMillis());

    public RPCSession(RPCServer server, String sysId, ChannelHandlerContext ctx) {
        this.server = server;
        this.sysId = sysId;
        this.ctx = ctx;
    }

    public String getSysId() {
        return sysId;
    }

    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) ctx.channel().remoteAddress();
    }

    public boolean isConnected() {
        return ctx.channel().isOpen();
    }

    public boolean handle(ByteBuf msg) {
        int len = msg.readableBytes();
        if (len > 4) {
            int msgLen = msg.getInt(msg.readerIndex());
            if (msgLen <= 0 || msgLen > 8 * 1024 * 1024) {
                throw new RuntimeException("协议错误，消息长度:" + msgLen);
            }
            if (len >= msgLen + 4) {
                byte[] data = new byte[msgLen];
                msg.readInt();
                msg.readBytes(data);
                addData(data);
                return handle(msg);
            }
        }
        return msg.readableBytes() == 0;
    }

    private void addData(final byte[] str) {
        if (msgIdBuf == null) {
            msgIdBuf = str;
        } else if (clzNameBuf == null) {
            clzNameBuf = str;
        } else {
            byte[] msgId = msgIdBuf;
            byte[] clzName = clzNameBuf;
            server.getThreadPool().submit(() -> handler(msgId, clzName, str));
            msgIdBuf = null;
            clzNameBuf = null;
        }
    }

    public RPCResponse invoke(final RPCRequest req) {
        String msgId = "S_" + idGen.incrementAndGet();
        RPCResponse response = null;
        responseMap.put(msgId, null);
        String lock = msgId.intern();
        synchronized (lock) {
            writeData(msgId, req);
            try {
                lock.wait(1200000);
            } catch (InterruptedException e) {
            }
            response = responseMap.remove(msgId);
        }
        if (response == null) {
            response = new RPCResponse();
            response.setError(true);
            response.setErrorMsg("请求超时");
        }
        return response;
    }

    private Map<String, String> responseStrMap = new HashMap<String, String>();

    public String invoke(String requestClzName, String requestStr) {
        String msgId = "H_" + idGen.incrementAndGet();
        String response = null;
        responseStrMap.put(msgId, null);
        String lock = msgId.intern();
        synchronized (lock) {
            try {
                _writeData(msgId.getBytes(), requestClzName.getBytes("UTF-8"), requestStr.getBytes("UTF-8"));
            } catch (UnsupportedEncodingException e1) {
            }
            try {
                lock.wait(120000);
            } catch (InterruptedException e) {
            }
            response = responseStrMap.remove(msgId);
        }
        if (response == null) {
            response = "请求超时";
        }
        return response;
    }

    private void handler(byte[] msgIdB, byte[] clzNameB, byte[] dataB) {
        Class<?> clz = null;
        String clzName = null;
        String data = null;
        try {
            clzName = new String(clzNameB, "UTF-8");
            String msgId = new String(msgIdB, "UTF-8");
            if (ZipUtil.isCompress(dataB)) {
                dataB = ZipUtil.uncompress(dataB);
            }
            data = new String(dataB, "UTF-8");
            if (msgId.startsWith("H_")) {
                String lock = msgId.intern();
                synchronized (lock) {
                    if (responseStrMap.containsKey(msgId)) {
                        responseStrMap.put(msgId, data);
                        lock.notify();
                    }
                }
                return;
            }
            clz = RPCUtil.getClassLoader().loadClass(clzName);
            Object msg = JSONValue.parse(data, clz);
            if (msg instanceof RPCRequest) {
                RPCRequest req = (RPCRequest) msg;
                Map<String, RPCHandler> handlerMap = server.getRPCHandlerMap().get(req.getInterfaceName());
                if (handlerMap == null) {
                    returnErrorMsg("不能处理的请求(" + req.getInterfaceName() + "::" + req.getMethodName() + ")", msgId);
                } else {
                    RPCHandler handler = handlerMap.get(req.getMethodName());
                    if (handler == null) {
                        returnErrorMsg("不能处理的请求(" + req.getInterfaceName() + "::" + req.getMethodName() + ")", msgId);
                    } else {
                        RPCContext.setRPCSession(this);
                        try {
                            RPCResponse response = handler.handler((RPCRequest) msg);
                            if (response != null) {
                                writeData(msgId, response);
                            } else {
                                response = new RPCResponse();
                                writeData(msgId, response);
                            }
                        } catch (Throwable t) {
                            returnErrorMsg(
                                    "请求处理错误(" + req.getInterfaceName() + "::" + req.getMethodName() + ","
                                            + ExceptionUtils.getRootCauseMessage(t) + ")", msgId);
                        }
                    }
                }
            } else if (msg instanceof RPCResponse) {
                RPCResponse response = (RPCResponse) msg;
                String lock = msgId.intern();
                synchronized (lock) {
                    if (responseMap.containsKey(msgId)) {
                        responseMap.put(msgId, response);
                        lock.notify();
                    }
                }
            }
        } catch (Throwable t) {
            t.printStackTrace();
            // TODO logger.error("处理请求错误", t);
        } finally {
            RPCContext.setRPCSession(null);
        }
    }

    private void writeData(String msgId, Object msg) {
        String str = JSONValue.toJSONString(msg);
        String clzName = msg.getClass().getName();
        byte[] msgIdData = toUTF(msgId);
        byte[] clzNameData = toUTF(clzName);
        byte[] data = toUTF(str);
        _writeData(msgIdData, clzNameData, data);
    }

    private synchronized void _writeData(byte[] msgIdData, byte[] clzNameData, byte[] data) {
        ByteBuf buffer = Unpooled.buffer(4 + msgIdData.length + 4 + clzNameData.length + 4 + data.length);
        buffer.writeInt(msgIdData.length);
        buffer.writeBytes(msgIdData);
        buffer.writeInt(clzNameData.length);
        buffer.writeBytes(clzNameData);
        if (data.length >= Constants.MIN_COMPRESS_LENGTH) {
            data = ZipUtil.compress(data);
            buffer.writeInt(data.length + RPCConstants.compressHeader.length);
            buffer.writeBytes(RPCConstants.compressHeader);
            buffer.writeBytes(data);
        } else {
            buffer.writeInt(data.length);
            buffer.writeBytes(data);
        }
        ctx.writeAndFlush(buffer);
    }

    private byte[] toUTF(String str) {
        byte[] data = null;
        try {
            data = str.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
        }
        return data;
    }

    private void returnErrorMsg(String msg, String msgId) {
        RPCResponse response = new RPCResponse();
        response.setError(true);
        response.setErrorMsg(msg);
        writeData(msgId, response);
    }
}